package com.sunzm.flink.datastream.scala.operator

import org.apache.flink.streaming.api.scala._
import org.slf4j.{Logger, LoggerFactory}

object ScalaMaxByMinByDemo {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {
    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    val dataStream: DataStream[String] = env.fromElements("zs,17,北京", "zs,20,上海", "zs,16,深圳")

    val keyedDS: KeyedStream[(String, Int, String), String] = dataStream.map(line => {
      val fileds: Array[String] = line.split(",")

      (fileds(0), fileds(1).toInt, fileds(2))
    }).keyBy(_._1)

    val maxAgeDS: DataStream[(String, Int, String)] = keyedDS.max(1)
    val maxByAgeDS: DataStream[(String, Int, String)] = keyedDS.maxBy(1)

    maxAgeDS.print("max")
    maxByAgeDS.print("maxBy")

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }
}
